package queue

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type DelayQueue[T Delayable] struct {
	q             *PriorityQueue[T]
	mutex         *sync.Mutex
	dequeueSignal *cond
	enqueueSignal *cond
}

func NewDelayQueue[T Delayable](capacity int) *DelayQueue[T] {
	m := &sync.Mutex{}
	return &DelayQueue[T]{
		q: NewPriorityQueue[T](capacity, func(src, dst T) int {
			srcDelay := src.Delay()
			dstDelay := dst.Delay()
			if srcDelay < dstDelay {
				return -1
			} else if srcDelay == dstDelay {
				return 0
			} else {
				return 1
			}
		}),
		mutex:         m,
		dequeueSignal: newCond(m),
		enqueueSignal: newCond(m),
	}
}

// 入队和并发阻塞队列没太大区别
func (c *DelayQueue[T]) EnQueue(ctx context.Context, data T) error {
	var err error
	for {
		//超时
		if ctx.Err() != nil {
			return ctx.Err()
		}
		c.mutex.Lock()
		if ctx.Err() != nil {
			return ctx.Err()
		}
		err = c.q.Enqueue(data)
		switch err {
		case nil:
			//入队成功
			c.enqueueSignal.broadcast()
			return nil
		case ErrOutOfCapacity:
			//入队失败 要等信号
			signal := c.dequeueSignal.signalCh()
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-signal:

			}
		default:
			//未知错误
			c.mutex.Unlock()
			return fmt.Errorf("ekit: 延时队列入队的时候遇到未知错误 %w，请上报", err)
		}
	}
}

// 出队就有讲究了：
//  1. Delay() 返回 <= 0 的时候才能出队
//  2. 如果队首的 Delay()=300ms >0，要是 sleep，等待 Delay() 降下去
//  3. 如果正在 sleep 的过程，有新元素来了，
//     并且 Dealay() = 200 比你正在sleep 的时间还要短，你要调整你的 sleep 时间
//  4. 如果 sleep 的时间还没到，就超时了，那么就返回
//
// sleep 本质上是阻塞（你可以用 time.Sleep，你也可以用 channel）
func (c *DelayQueue[T]) DeQueue(ctx context.Context) (T, error) {
	var t T

	var timer *time.Timer

	defer func() {
		if timer != nil {
			timer.Stop()
		}
	}()

	for {
		//超时
		if ctx.Err() != nil {
			return t, ctx.Err()
		}
		c.mutex.Lock()
		//超时
		if ctx.Err() != nil {
			return t, ctx.Err()
		}
		t, err := c.q.Peek()
		switch err {
		case nil:
			//获取成功
			delay := t.Delay()
			if delay <= 0 {
				// 可以出队
				t, err = c.q.Dequeue()
				c.dequeueSignal.broadcast()
				return t, err
			}
			if timer == nil {
				timer = time.NewTimer(delay)
			} else {
				timer.Reset(delay)
			}
			signal := c.enqueueSignal.signalCh()
			select {
			case <-timer.C:
				//时间到，可以出队
			case <-ctx.Done():
				//超时
				return t, ctx.Err()
			case <-signal:
			}

		case ErrEmptyQueue:
			//空 等待信号唤醒
			signal := c.enqueueSignal.signalCh()
			select {
			case <-ctx.Done():
				return t, ctx.Err()
			case <-signal:

			}

		default:
			//未知错误
			c.mutex.Unlock()
			return t, fmt.Errorf("ekit: 延时队列出队的时候遇到未知错误 %w，请上报", err)

		}

	}
}

type Delayable interface {
	Delay() time.Duration
	// Deadline() time.Time
}

type cond struct {
	signal chan struct{}
	l      sync.Locker
}

func newCond(l sync.Locker) *cond {
	return &cond{
		signal: make(chan struct{}),
		l:      l,
	}
}

// broadcast 唤醒等待者
// 如果没有人等待，那么什么也不会发生
// 必须加锁之后才能调用这个方法
// 广播之后锁会被释放，这也是为了确保用户必然是在锁范围内调用的
func (c *cond) broadcast() {
	signal := make(chan struct{})
	old := c.signal
	c.signal = signal
	c.l.Unlock()
	close(old)
}

// signalCh 返回一个 channel，用于监听广播信号
// 必须在锁范围内使用
// 调用后，锁会被释放，这也是为了确保用户必然是在锁范围内调用的
func (c *cond) signalCh() <-chan struct{} {
	res := c.signal
	c.l.Unlock()
	return res
}
